本次的程式碼與目錄結構可以參考 FastAPI Tutorial : Day25 branch
在 昨天 我們完成 Redis 的基本操作包括:
Redis
類別來建立連線ConnectionPool
來管理連線redis-py
提供的 set
、 save
redis-om
的 Object Mapping 來做 CRUD有這些基本的認識後
今天可以嘗試使用 redis-om
來實作 Server Cache
並比較 fastapi-cache
和 redis-py
來實作 Cache 的差異
這次想實作的 Server Cache 是在 CRUD 層面的 Cache
而不是在 Response Level 的 Cache
所以在遇到 Read
操作時,我們會先去 Cache 中尋找資料
如果 Cache 中有資料,則直接回傳
並把 user_id
當作 key
來存取
而 Create
, Update
, Delete
操作時
我們會依據 選定的 Key 來更新或刪除 key-value pair
fastapi-cache
實作因為該套件主要是針對 Response Level 的 Cache
是透過 decorator 來 Cache 我們要 Response 前的結果
用法大略如下:
from fastapi_cache.decorator import cache
@app.get("/users/{user_id}")
@cache(expire=60) # <----- 60 秒後過期
async def read_user(user_id: int):
return {"user_id": user_id}
但是 fastapi-cache
無法做到遇到:
「 Create
, Update
, Delete
操作時,要依需相對應的 Key 來更新刪除 Key-value Pair 」的操作
redis-om
實作如果由 redis-om
來實作的話
我們也需要先定義 UserCacheModel
class UserCache( JsonModel ):
id: Optional[int] = Field(index=True)
name:Optional[str] = Field(index=True)
password:Optional[str] = Field(index=False)
name: Optional[str] = Field(index=True)
avatar: Optional[str] = Field(index=False)
age: Optional[int] = Field(index=False,default=0)
email: Optional[EmailStr] = Field(index=True)
birthday: Optional[date] = Field(index=False)
class Meta:
database = redis
為了盡量不改動到 CRUD 的 Code
我們可以使用 decorator 來實作
又因為讓 cache decorator 保留彈性,我們可以傳入 key
來當作 目前應該要依據哪個欄位搜尋 cls
則是 CRUD 回傳的 Schema
database/redis_cache.py
def generic_cache_get(key:str,cls):
def inner(func):
async def wrapper(*args, **kwargs):
# TODO
return wrapper
return inner
再將內層 get 、 set 邏輯實作完 database/redis_cache.py
try:
redis_result = UserCache.find( ModelField(name=key,type=type(value_key)) == value_key ).first()
except:
redis_result = None
if redis_result:
return cls(**redis_result.dict())
else:
result = await func(*args, **kwargs)
UserCache(**result.dict()).save()
return result
redis-om
內建 Model 實作的坑會發現在使用上,每 Model.save()
一次後
在 redis
中就會多一個 key
假設我們要以 user_id
當 Key 的 Cache 資料只有 id
, name
, email
, avatar
四個欄位
另一個以同樣以 user_id
最為 Key 的 Cache 資料只有 id
, password
兩個欄位
還需要我們額外處理
又因為 redis-om
的 Model 是透過 pydantic
來實作
像是 email 綁定 EmailStr
、 birthday 綁定 date
會導致一些 Validation 的問題
因為 redis-om
綁定 pydantic
的關係
沒有辦法使用的這麼彈性
redis-om
結果觀察 redis-om
的結果可以發現
是將 modulename + classname + pk 當作 key
如下所示:
( HashModel
生成的 Key 格式 )
( JsonModel
生成的 Key 格式 )
redis-py
和自定義 Model 實作既然使用 redis-om
會有額外的問題
也須要額外處理
那不如我們就直接使用 redis-py
來實作
同樣也是使用 decorator 來實作
Cache 主要分成 3 個操作
並使用 Redis 內建的 Hash
Type 來實作
redis_connection.hgetall( key_name )
: 回傳 key_name
下所有的 key-value pair ( dict )redis_connection.hset( key_name , mapping )
: 將 mapping
中的 key-value pair 寫入 key_name
中 ( 可以當作新增或更新 )又考慮到複用性 ,我們將 perfix
和 key
作為參數傳入
讓能夠以 prefix + kargs[key] 作為 Key
所以要被 Wrap 起來的 function 必須要有
**kwargs
來傳參數
database/redis_cache.py
def generic_cache_get(prefix:str,key:str,cls:object):
'''
prefix: namspace for redis key ( such as `user` 、`item` 、`article` )
key: **parameter name** in caller function ( such as `user_id` 、`email` 、`item_id` )
cls: **response schema** in caller function ( such as `UserSchema.UserRead` 、`UserSchema.UserId` 、`ItemSchema.ItemRead` )
'''
rc = redis.Redis(connection_pool=redis_pool)
def inner(func):
async def wrapper(*args, **kwargs):
# TODO ...
return wrapper
return inner
如果有 cache_key
還需要檢查目前 Cache 的結果是否有所有 Schema 包含的 Field
如果有缺少 Field 話
也需要先去 DB 中取得資料後再更新!
database/redis_cache.py
#...
# 在呼叫被包裝的 function 時,要以 func(key=value) 的方式傳入參數
value_key = kwargs.get(key)
if not value_key:
return await func(*args, **kwargs)
# 組合成給 redis 的 key
cache_key = f"{prefix}:{value_key}"
# 檢查是否有 cache
try:
redis_result:dict = rc.hgetall(cache_key)
# 即使有結果 還需要檢查是否有所有的 Field
if check_has_all_keys(redis_result,cls): # cache hit !
return cls(**redis_result)
except:
pass
sql_result = await func(*args, **kwargs)
if not sql_result:
return None
rc.hset(cache_key, mapping=sql_query_row_to_dict(sql_result))
return sql_result
我們這邊直接用 exception 來判斷是否有 cache hit
而不用 redis_connect.exists
來檢查來加速
在 CRUD 中,我們只需要加上 @generic_cache_get
來使用 crud/users.py
# ...
@generic_cache_get(prefix="user",key="user_id",cls=UserSchema.UserRead)
async def get_user_infor_by_id(self,user_id:int ,db_session:AsyncSession) -> UserSchema.UserInfor:
# ...
@generic_cache_get(prefix="user",key="user_id",cls=UserSchema.UserId)
async def get_user_id_by_id(self,user_id:int ,db_session:AsyncSession=None) -> UserSchema.UserId:
# ...
@generic_cache_get(prefix="user",key="email",cls=UserSchema.UserInDB)
async def get_user_in_db(self,email:str , db_session:AsyncSession=None) -> UserSchema.UserInDB :
# ...
# ...
要注意的是
當我們在 call 這些 CRUD function 時,要以 func(key=value)
的方式傳入參數
api/users.py
# ...
# ...
async def get_user_infor_by_id(user_id: int):
# 這邊要以 `user_id=user_id` 的方式傳入
user = await UserCrud.get_user_infor_by_id(user_id=user_id)
api/auth.py
async def login(form_data: login_form_schema): # ...
# 要以 `email=form_data.username` 的方式傳入
user_in_db:UserInDB = await UserCrud.get_user_in_db(email=form_data.username)
對於 Update 和 Delete Cache 的操作也與 Get 類似
也是先從 DB 中取得資料後以 hset
更新
Delete 則是直接依據 key
來刪除
這邊就不佔篇幅
可以直接看 Day25 branch: database/redis_cache.py
這邊都以 100 個 user 和隨機的合法 user_id 來測試
原本的
加上 Redis Cache 後
可以看到大約快了 12 秒 !
原本的
加上 Redis Cache 後
可以看到大約快了 150 秒 !
今天我們使用 redis-py
來實作 Server Cache
也看到以 redis-om
來實作的坑
最後以 Redis 內建的 Hash
Type 來實作
包裝成可複用的 decorator
使用 Redis 作為 Server Cache 也有看到不錯的效果
明天會接著把 pagenation ( get_user_list
CRUD ) 也加入 Cache